package each_webcode

import common.HbaseTools
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import scala.collection.mutable
import scala.collection.mutable.HashMap
import scala.io.Source
import scala.util.parsing.json.JSON;
object transferValueCode_to_Name_UIS {

        def main(args: Array[String]): Unit = {
                //mongodb中的列
                //_id COUNTRY_ID HeadCode INDJCATOR_ID MAGNITUDE QUALIFITER SourceCode Unit VALUE Value ValueStr YEAR
                //KBase中的列
                val sparkConf = new SparkConf().setAppName("uisdatachange")//.setMaster("local[*]")
                val sc = new SparkContext(sparkConf)
                val baseInfotable: String = new String("BaseMapInfos") //DetailMapInfos
                val detailTable: String = new String("DetailMapInfos") //BaseMapInfos

                val inpath = args(0)  //"hdfs://node01:8020/tmp/InternationalData_source/InternationalData.UIS_Value.json"
                val outpath = args(1) //"hdfs://node01:8020/tmp/UIS_TEST/result"


                //val inpath ="D:\\InternationalData\\makeChange\\src\\main\\resources\\sub_UIS_Value.json"
                //val outpath = "hdfs://node01:8020//tmp/InternationalData_tmp/sub_uis_test_14"

                val webcodes = List("UIS")
                //make a cache:
                        val inputFile = inpath //读取json文件
                val jsonStr = sc.textFile(inputFile) //.sample(true,0.01);

                val result = jsonStr.map(s => JSON.parseFull(s))

                val rdds = result.map {
                        r =>
                                r match {
                                        case Some(maps: Map[String, Any]) => {
                                                val webcode = "UIS"
                                                val thashmap: HashMap[String, String] = new HashMap[String, String]()
                                                maps.foreach(t => {
                                                        thashmap += (t._1.toString -> t._2.toString.replace("\n",""))
                                                        })
                                                //maybe there should deal with _id
                                                thashmap("_id") = thashmap("_id").substring(12, 36)
                                                // get webcode + sourcecode
                                                // rowkeys
                                                //随机数____SourceCode
                                                (scala.util.Random.nextInt(20).toString + "____"+ thashmap("SourceCode"), thashmap)
                                                //(webcode + "_" + thashmap("SourceCode"), thashmap)
                                                //(scala.util.Random.nextInt(10).toString + "_" + webcode + "_" + thashmap("SourceCode"), thashmap)
                                                //("0"+"_WTO"+"_"+thashmap("SourceCode"),thashmap)
                                                }
                                        }
                        }.groupByKey().map {
                        case (k, v) => {
                                val k1array = k.split("____")
                                //val k1 = k1array(1) + "_" + k1array(2)
                                        //UIS_SourceCode
                                        val k1 = "UIS" + "_" + k1array(1)

                                //by k,you will get the mapinfos
                                        val tools = new HbaseTools()
                                val connection = tools.getConnection()
                                val dealTypes: Map[String, String] = tools.getFamily_column_value(connection,baseInfotable, k1, "dealTypes").toMap
                                val map01: Map[String, String] = tools.getFamily_column_value(connection,baseInfotable, k1, "map01").toMap
                                val map02: Map[String, String] = tools.getFamily_column_value(connection,baseInfotable, k1, "map02").toMap
                                val webinfo:Map[String, String] = tools.getFamily_column_value(connection,baseInfotable, k1, "WebInfo").toMap
                                val PU_timeMap:Map[String,String] = tools.getFamily_column_value(connection,baseInfotable,"public","public_Time").toMap
                                val PU_country:Map[String,String] = tools.getFamily_column_value(connection,baseInfotable,"public","public_country").toMap
                                //create a map for map04
                                        //then for each v;you can transfer code to value
                                        //"DetailMapInfos,"UIS_uis"的所有列族
                                        val map04: HashMap[String, Map[String, String]] = new HashMap[String, Map[String, String]]()
                                val myfamilys = tools.getFamilys_list(connection,detailTable, "UIS_uis")
                                for (onefamily: String <- myfamilys) {
                                        val tmp = tools.getFamily_column_value(connection,detailTable, "UIS_uis", onefamily).toMap
                                        map04.put(onefamily, tmp)
                                        }

                                connection.close()
                                //构建
                                        val onekeysValue = v.map {
                                        detailValuecode => { // get one record
                                                var oneLinceMap:HashMap[String,String] = new mutable.HashMap[String,String]()
                                                //map03
                                                        var map03 = detailValuecode
                                                dealTypes.keys.foreach {
                                                        keys =>
                                                                dealTypes(keys) match {

                                                                        case "type01" => {
                                                                                //map01 + map03
                                                                                if (map01.contains(keys) && map03.contains(map01(keys))) {
                                                                                        oneLinceMap.put(keys, map03(map01(keys)))
                                                                                        }
                                                                                }
                                                                                case "type02" => {
                                                                                        //map01 + map03 + map04
                                                                                        oneLinceMap.put(keys, webinfo((map01(keys))))
                                                                                        }
                                                                                case "type03" => {
                                                                                        /*map01 + map02 =conceptid
                                                                                       map01 + map03 =detailId
                                                                                       conceptid + detailid + map04
                                                                                      */
                                                                                        if (map03(map01(keys)) != "") {
                                                                                                oneLinceMap.put(keys,map04(map02(map01(keys))).getOrElse(map03(map01(keys)),""))
                                                                                                }
                                                                                        }
                                                                                case "type04" => {
                                                                                        //map01 + map02 = conceptId  MAGNITUDE
                                                                                                var qual = ""
                                                                                        if(map03("QUALIFIER")==""){
                                                                                                qual = ""
                                                                                                }else{
                                                                                                if(map04("QUALIFIER").contains(map03("QUALIFIER"))){
                                                                                                        qual = map04("QUALIFIER")(map03("QUALIFIER"))
                                                                                                        }
                                                                                                }
                                                                                        var mag = ""
                                                                                        if(map03("MAGNITUDE")==""){
                                                                                                mag = ""
                                                                                                }else{
                                                                                                if(map04("MAGNITUDE").contains(map03("MAGNITUDE"))){
                                                                                                        mag = map04("MAGNITUDE")(map03("MAGNITUDE"))
                                                                                                        }
                                                                                                }
                                                                                        var resul = ""
                                                                                        if(qual=="" && mag==""){}
                                                                                                else if(qual=="" && mag!=""){
                                                                                                resul = qual
                                                                                                }
                                                                                                else if(qual!="" && mag==""){
                                                                                                resul = "["+mag+"]"
                                                                                                }
                                                                                                else{
                                                                                                resul = qual + "["+mag+"]"
                                                                                                }

                                                                                        oneLinceMap.put(keys,resul)
                                                                                        }
                                                                        }
                                                        }
                                                //合并字段，生成
                                                //检测 Year字段，
                                                if(oneLinceMap.contains("Year")&&oneLinceMap("Year")!=""){
                                                        if(oneLinceMap("Year").length>4 && !oneLinceMap.contains("Period")){
                                                                oneLinceMap.put("Period",oneLinceMap("Year").substring(4))
                                                                oneLinceMap("Year") = oneLinceMap("Year").substring(0,4)
                                                                }
                                                        //检测Period字段
                                                        //1检查是否有值，
                                                        }
                                                //生成TimeSort字段；Period字段判断是否存在并进行转换，
                                                if(oneLinceMap.contains("Year")&&oneLinceMap.contains("Period")&&oneLinceMap("Period").trim!=""){
                                                        if(PU_timeMap.contains(oneLinceMap("Period").trim)){
                                                                oneLinceMap.put("TimeSort",oneLinceMap("Year")+PU_timeMap(oneLinceMap("Period").trim))
                                                                }else{
                                                                oneLinceMap.put("TimeSort",oneLinceMap.getOrElse("Year",""))
                                                                }
                                                        }else{
                                                        oneLinceMap.put("TimeSort",oneLinceMap.getOrElse("Year",""))
                                                        }
                                                //生成FuzzyIndicator字段
                                                if(oneLinceMap.contains("Indicator")&&oneLinceMap("Indicator")!="")
                                                        {oneLinceMap.put("FuzzyIndicator",oneLinceMap("Indicator"))
                                                                //生成IndicatorLength字段
                                                                oneLinceMap.put("IndicatorLength",oneLinceMap("Indicator").length.toString)
                                                                }
                                                //生成FuzzyReporter字段
                                                if(oneLinceMap.contains("Reporter")&&oneLinceMap("Reporter")!=""){
                                                        if(PU_country.contains(oneLinceMap("Reporter"))){
                                                                oneLinceMap.put("FuzzyReporter",oneLinceMap("Reporter")+"，"+PU_country(oneLinceMap("Reporter")))
                                                                }else{
                                                                oneLinceMap.put("FuzzyReporter",oneLinceMap("Reporter"))
                                                                }
                                                        }
                                                //Parterner
                                                //生成FUZZYPARTERNER字段
                                                if(oneLinceMap.contains("Parterner")&&oneLinceMap("Parterner")!=""){
                                                        if(PU_country.contains(oneLinceMap("Parterner"))){
                                                                oneLinceMap.put("FUZZYPARTERNER",oneLinceMap("Parterner")+"，"+PU_country(oneLinceMap("Parterner")))
                                                                }else{
                                                                oneLinceMap.put("FUZZYPARTERNER",oneLinceMap("Parterner"))
                                                                }
                                                        }
                                                val MetaData:Array[String] = Array("WebCode", "SourceCode", "ReporterCode", "Reporter", "ParternerCode", "Parterner", "Year", "Period",
                                                        "IndicatorCategory", "Indicator", "Dim1Type", "Dim1", "Dim2Type", "Dim2", "Dim3Type", "Dim3",
                                                        "Dim4Type", "Dim4", "Dim5Type", "Dim5", "Dim6Type", "Dim6", "Dim7Type", "Dim7", "Dim8Type", "Dim8" )
                                                val DIMTYPEMETADATA:Array[String] =Array("Dim1Type", "Dim2Type", "Dim3Type", "Dim4Type", "Dim5Type", "Dim6Type", "Dim7Type", "Dim8Type")
                                                val DIMMETADATA:Array[String] = Array("Dim1", "Dim2", "Dim3", "Dim4", "Dim5", "Dim6", "Dim7", "Dim8")

                                                //生成DIMTYPEMETADATA
                                                        var dimtypemetadata = DIMTYPEMETADATA.flatMap(oneLinceMap.getOrElse(_,"<->")+"，").mkString("").replace("<->，","")

                                                //生成DIMMETADATA
                                                        var dimemetadata = DIMMETADATA.flatMap(
                                                        onecel=>{
                                                                oneLinceMap.getOrElse(onecel,"<->")+"，"
                                                                }
                                                        ).mkString("").replace("<->，","")
                                                //生成MetaData
                                                        var metadata = MetaData.flatMap(oneLinceMap.getOrElse(_,"<->")+"，").mkString("").replace("<->，","")
                                                oneLinceMap.put("DIMTYPEMETADATA",dimtypemetadata)
                                                oneLinceMap.put("DIMMETADATA",dimemetadata)
                                                oneLinceMap.put("MetaData",metadata)
                                                //val kkl:RDD[String] = sc.parallelize(().toString)
                                                oneLinceMap.mkString("(<>)").replace(" -> ","\t")
                                                }
                                        }
                                var fi = onekeysValue.mkString( "\n")
                                fi
                                }
                        }.saveAsTextFile(outpath)
                sc.stop()
                }
        }
